package com.mqy.service;
/**
 * This file created by mengqingyi on 2018/7/9.
 */

import com.google.common.collect.Lists;

import com.mqy.config.PhoenixConnectionByAliPool;
import com.mqy.config.PhoenixConnectionByJDBC;
import com.mqy.dao.HBaseUserLocationDao;
import com.mqy.entity.Page;
import com.mqy.entity.UserLocation;
import com.mqy.utils.HbaseStringFilter;

import org.apache.phoenix.jdbc.PhoenixConnectionPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Repository;

import java.sql.Connection;
import java.sql.Date;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.text.SimpleDateFormat;
import java.util.List;
import java.util.TimeZone;

/**
 * @author mengqingyi
 * @classDescription 用户地理位置信息加盐表改造  使用原生JDBC连接来进行查询操作 原先查询进行汇总拆分
 * @create 2018-07-09 15:04
 *
 * 目前使用到的SQL
 *
 * #userLocation表
 *
 * select userId,address,lng,lat,createTime as createTime2 from userlocation where userid=? order by createTime desc ;
 *
 * select userId,address,lng,lat,createTime as createTime2 from userlocation where where userId != ? and address
 * in(?,?,?);
 *
 * select userId,address,lng,lat,createTime as createTime2  from userlocation where userId = ? order by createTime  desc
 * limit 800;
 *
 * upsert into  userlocation values (?,?,?,?,?,?);
 *
 * select address,createTime as createTime2 from userlocation where userId = ? order by createTime  desc  limit 1;
 *
 * select count(1) from userlocation where userId = ?;
 *
 * select userId,createTime,lng,lat from userlocation where userId =? and createTime <=to_date(?) order by createTime
 * desc limit 1;
 **/
@Repository
public class HbaseSaltUserLocationDaoImpl implements HBaseUserLocationDao {
    /**
     * 日志
     */
    //private static final Logger logger = LoggerFactory.getLogger(HbaseSaltUserLocationDaoImpl.class);
    /**
     * 获取连接 4种类 方式
     */
    //其一  需要注意的是 共享connection 不可直接关闭
    // @Autowired
    // @Qualifier("getDefaultConnection")
    // private Connection connection;
    //其二 使用阿里的 PhoenixConnectionPool 关闭连接扔回队列
    // @Autowired
    // @Qualifier("getAliPhoenixConnectionPool")
    // private PhoenixConnectionPool aliPhoenixConnectionPool;
    //其三 通过静态方法
    //Connection connection = PhoenixConnectionByJDBC.getConnection();
    //其四 连接池 BasicDataSource+ spring jdbcTemplete
    // @Autowired
    // @Qualifier("phoenixJdbcTemplate")
    // private JdbcTemplate jdbcTemplate;


    /**
     * 查询字段
     */
    private final static String SALT_USERLOCATION_QUERY_SQL = "SELECT USERID,ADDRESS,LNG,LAT,CREATETIME FROM " +
            "SALT_USER_LOCATION ";
    /**
     * 保存
     */
    private final static String SALT_USERLOCATION_UPSERT_SQL = "UPSERT INTO SALT_USER_LOCATION(USERID,ADDRESS," +
            "LNG,LAT,CREATETIME)  VALUES (?,?,?,?,?)";
    /**
     * 如果语句中没有 limit  那么默认只查询800 条
     */
    private final static String LIMIT_SQL = " LIMIT 800 ";
    /**
     * limit关键字
     */
    private final static String LIMIT_KEYWORD = "LIMIT";


    /**
     * 基础查询类  下面所有的实现方法通过调用该方法去实现
     *
     * 其他方法相当于主要任务是做 where sql 语句的拼接
     */
    @Override
    public List<UserLocation> query(String whereSql) {
        long start = System.currentTimeMillis();
        Connection connection = PhoenixConnectionByJDBC.getConnection();
        // Connection connection = null;
        // try {
        //     connection = aliPhoenixConnectionPool.getConnection();
        // } catch (SQLException e) {
        //     //logger.error("[hbase salt_user_location]基础查询获取连接失败!!!" + e.getMessage());
        //     e.printStackTrace();
        // }
        PreparedStatement statement = null;
        ResultSet resultSet = null;
        List<UserLocation> userLocationList = Lists.newArrayList();
        String conditionSql = whereSql.toUpperCase();
        if (!conditionSql.contains(LIMIT_KEYWORD)) {
            conditionSql += LIMIT_SQL;
        }
        String psSql = SALT_USERLOCATION_QUERY_SQL + conditionSql;
        try {
            statement = connection.prepareStatement(psSql);
            resultSet = statement.executeQuery();
            while (resultSet.next()) {
                UserLocation userLocation = new UserLocation();
                long userid = resultSet.getLong("USERID");
                userLocation.setUserId(userid);
                String address = resultSet.getString("ADDRESS");
                userLocation.setAddress(address);
                Double lng = resultSet.getDouble("LNG");
                userLocation.setLng(lng);
                Double lat = resultSet.getDouble("LAT");
                userLocation.setLat(lat);
                //兼容以前代码 所以这有两个createTime
                Date createtime = resultSet.getDate("CREATETIME");
                String createtime2 = resultSet.getString("CREATETIME");
                userLocation.setCreateTime(createtime);
                userLocation.setCreateTime2(createtime2);
                userLocationList.add(userLocation);
            }
        } catch (Exception e) {
            //logger.error("[hbase salt_user_location]基础查询严重错误!!!" + e.getMessage());
            e.printStackTrace();
        } finally {
            Boolean close = PhoenixConnectionByAliPool.close(connection, statement, resultSet);
            if (!close) {
                //logger.error("[hbase salt_user_location]基础查询资源关闭异常!!!");
            }
        }
        long end = System.currentTimeMillis();
        //logger.info("[hbase salt_user_location]查询耗时{}ms", (end - start));
        return userLocationList;
    }

    /**
     * 拼接sql
     *
     * @param userLocation userLocation 实体类
     * @param sort         true 倒序排 flase正序排
     */
    @Override
    public List<UserLocation> query(UserLocation userLocation, boolean sort) {
        List<UserLocation> userLocationList = Lists.newArrayList();
        if (userLocation == null || userLocation.getUserId() == null) {
            return userLocationList;
        }
        String whereSql = " WHERE USERID = " + userLocation.getUserId();
        if (sort) {
            //由于几处调用均是以 createTime 字段排序 所以将这个字段直接写死
            whereSql += "  ORDER BY  CREATETIME  DESC ";
        }
        long start = System.currentTimeMillis();
        userLocationList = query(whereSql);
        long end = System.currentTimeMillis();
        //logger.info("[hbase salt_user_location]query分页查询拼接whereSql{},耗时{}ms", whereSql, (end - start));
        return userLocationList;
    }


    /**
     * 分页查询
     *
     * @param userLocation 实体类
     * @param page         分页参数
     */
    @Override
    public Page<UserLocation> query(UserLocation userLocation, Page<UserLocation> page) {
        Long userId = userLocation.getUserId();
        if (page == null || userId == null) {
            //logger.error("[hbase salt_user_location]入参存在空值");
            return page;
        }
        //查询总记录数
        Long count = count(userLocation);
        StringBuilder builder = new StringBuilder();
        builder.append(" WHERE USERID = ").append(userId);
        if (page.getOrderBy() != null) {
            builder.append(" ORDER BY ").append(page.getOrderBy());
            if (page.getOrderDir() != null) {
                builder.append(" ").append(page.getOrderDir());
            }
        }
        builder.append(" LIMIT ").append(page.getPageSize()).append(" OFFSET ").append(page.getPageNo());
        String whereSql = builder.toString();
        long start = System.currentTimeMillis();
        //查询结果集
        List<UserLocation> query = query(whereSql);
        page.setTotalCount(count);
        page.setResult(query);
        long end = System.currentTimeMillis();
        //logger.info("[hbase salt_user_location]query分页查询拼接whereSql{},耗时{}ms", whereSql, (end - start));
        return page;
    }

    /**
     * 保存地理位置信息
     *
     * @param userLocation 实体
     */
    @Override
    public void add(UserLocation userLocation) {
         Connection connection = PhoenixConnectionByJDBC.getConnection();
        // Connection connection = null;
        // try {
        //     connection = aliPhoenixConnectionPool.getConnection();
        // } catch (SQLException e) {
        //     //logger.error("[hbase salt_user_location]基础查询获取连接失败!!!" + e.getMessage());
        //     e.printStackTrace();
        // }
        PreparedStatement statement = null;
        if (userLocation == null) {
            //logger.error("【HBase】（userLocation）入参userLocation为NULL");
            return;
        }
        //phoenix时区问题  需要使用东八区 由于历史数据 通过程序转换 故还用程序进行时区转换
        // 如果是全新的 可以 通过 phoenix bin 下的 hbase-site.xml添加配置
            /*
             <property>
                <name>phoenix.query.dateFormatTimeZone</name>
                <value>Asia/Shanghai</value>
             </property>*/
        SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        simpleDateFormat.setTimeZone(TimeZone.getTimeZone("Asia/Shanghai"));
        try {
            long start = System.currentTimeMillis();
            statement = connection.prepareStatement(SALT_USERLOCATION_UPSERT_SQL);
            statement.setLong(1, userLocation.getUserId());
            statement.setString(2, HbaseStringFilter.stringFilter(userLocation.getAddress()) );
            statement.setDouble(3, userLocation.getLng());
            statement.setDouble(4, userLocation.getLat());
            statement.setString(5, simpleDateFormat.format(userLocation.getCreateTime()));
            int execute = statement.executeUpdate();
            connection.commit();
            long end = System.currentTimeMillis();
            //logger.info("[hbase salt_user_location]add保存方法影响行数{},耗时{}ms", execute, (end - start));
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            PhoenixConnectionByAliPool.close(connection, statement,null);
        }

    }

    /**
     * 查询用户最新的一条地理位置信息
     *
     * @param userLocation 实体
     */
    @Override
    public List<UserLocation> queryNewTimeAddress(UserLocation userLocation) {
        List<UserLocation> userLocationList = Lists.newArrayList();
        Long userId = userLocation.getUserId();
        if (userId == null) {
            return userLocationList;
        }
        String whereSql = " WHERE USERID = " + userId + "  ORDER BY  CREATETIME  DESC LIMIT 1 ";
        long start = System.currentTimeMillis();
        userLocationList = query(whereSql);
        long end = System.currentTimeMillis();
        //logger.info("[hbase salt_user_location]queryNewTimeAddress查询拼接whereSql{},耗时{}ms", whereSql, (end - start));
        return userLocationList;
    }

    /**
     * 查询该用户的记录条数
     *
     * @param userLocation 实体
     */
    @Override
    public Long count(UserLocation userLocation) {
        long start = System.currentTimeMillis();
        Connection connection = PhoenixConnectionByJDBC.getConnection();
        // Connection connection = null;
        // try {
        //     connection = aliPhoenixConnectionPool.getConnection();
        // } catch (SQLException e) {
        //     //logger.error("[hbase salt_user_location]基础查询获取连接失败!!!" + e.getMessage());
        //     e.printStackTrace();
        // }
        PreparedStatement statement = null;
        ResultSet resultSet = null;
        Long userId = userLocation.getUserId();
        Long count = 0L;
        if (userId == null) {
            //logger.error("[hbase salt_user_location]count入参存在空值,查询记录数直接置为0");
            return count;
        }
        String psSql = SALT_USERLOCATION_QUERY_SQL + " WHERE USERID=" + userId;
        // long start = System.currentTimeMillis();
        try {
            statement = connection.prepareStatement(psSql);
            resultSet = statement.executeQuery();
            while (resultSet.next()) {
                count++;
            }
        } catch (SQLException e) {
            //logger.error("[hbase salt_user_location]count记录数查询严重错误!!!" + e.getMessage());
            e.printStackTrace();
        } finally {
            Boolean close = PhoenixConnectionByAliPool.close(connection, statement, resultSet);
            if (!close) {
                //logger.error("[hbase salt_user_location]基础查询资源关闭异常!!!");
            }
        }
        long end = System.currentTimeMillis();
        //logger.info("[hbase salt_user_location]count查询耗时{}ms", (end - start));
        return count;
    }
}